package com.practice.car.cardataapp.schelduer.spark.task.recom;

import com.alibaba.fastjson.JSON;
import com.practice.car.cardataapp.entity.CarGrade;
import com.practice.car.cardataapp.entity.Subcar;
import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.RecomRunner;
import com.practice.car.cardataapp.service.SubCarService;
import com.practice.car.cardataapp.service.impl.SubCarServiceImpl;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import scala.Serializable;
import scala.Tuple2;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.List;

import static com.practice.car.cardataapp.schelduer.spark.Schedule.getConnection;

/**
 * car-bigdata project
 *
 * @author ZengHanxiong
 * @date 2020/9/13
 */
@Service
public class CarGradeAPP implements Serializable, RecomRunner {
    private Connection connection = null;
    @Autowired
    SubCarService subCarService;
    @Override
    @Test
    public void run() {
        String path =Schedule.recomHdfsUrl;//hdfs://120.24.6.0:9000/user/root/car_grade.json | Schedule.recomHdfsUrl
        //开启 SparkContext
        intitSubCar();//插入subcar数据
        JavaSparkContext sc = new JavaSparkContext("local[2]", "cargrade_application");
        JavaRDD<CarGrade> rdd = sc.parallelize(readJsonArray(path, CarGrade.class,""));
        makeGrade(rdd);
        sc.close();
    }

    public void intitSubCar() {
        String subCarUrl = Schedule.subCarUrl;
        List<Subcar> subcars = readJsonArray(subCarUrl, Subcar.class, "");
        System.out.println("subCarService = " + subCarService);
        new SubCarServiceImpl().saveOrUpdateBatch(subcars);
    }
    //车系平均评分
    public void makeGrade(JavaRDD<CarGrade> rdd){
        //Tuple2<Double:平均分,String:车系名>，对应改方法名就可以
        JavaPairRDD<String,Double> rddTuple = rdd.mapToPair(car -> new Tuple2<>(car.getName(),car))
                .groupByKey().mapToPair(car ->{
                    double sum = 0;
                    double count = 0;
                    Iterator<CarGrade> carIterator =car._2.iterator();
                    while(carIterator.hasNext()){
                        sum += carIterator.next().getGrade();
                        count++;
                    }
                    double avg = sum / count;
                    return new Tuple2<>(car._1(),((int)(avg*10))/10.0);
                });
        //Collection<Tuple2<Double,String>> collection = rddTuple.collect();
        rddTuple.sortByKey().foreachPartition(i -> i.forEachRemaining(e->{
            PreparedStatement preparedStatement = null;
            try {
                preparedStatement = getConnection().prepareStatement("update subcar set  grade= ?  where name=?");
                preparedStatement.setDouble(1, e._2);
                preparedStatement.setString(2, e._1);
                preparedStatement.executeUpdate();
            } catch (SQLException ex) {
                ex.printStackTrace();
            }
        }));
    }

    private <T> List<T> readJsonArray(String path, Class<T> clasz, String model) {//local模式，加载本地项目路径下的json文件，默认加载的是hdfs上的文件
        List<T> list = null;
        int available = 0;
        BufferedInputStream inputStream = null ;
        if ("local".equals(model)) {
            try {
                inputStream = new BufferedInputStream(new FileInputStream(path));
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }else{
            Path hdfsPath = new Path(path);
            Configuration configuration = new Configuration();
            try {
                FileSystem fileSystem = new Path(path).getFileSystem(configuration);
                inputStream= new BufferedInputStream(fileSystem.open(hdfsPath));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            available = inputStream.available();
            byte[] buffer = new byte[available];
            inputStream.read(buffer);
            inputStream.close();
            list = JSON.parseArray(new String(buffer, StandardCharsets.UTF_8), clasz);
        } catch (IOException e) {
            e.printStackTrace();
        }
        //将 text文件里的 json数据读取 List<Car>
        return list;
    }
}
